package com.storm.helloword;

import com.storm.helloword.bolt.ExclaimBasicBolt;
import com.storm.helloword.bolt.PrintBolt;
import com.storm.helloword.spout.RandomSpout;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.utils.Utils;
/**
 * 
  * @ClassName: ExclaimBasicTopology
  * @Description: 例子来源    http://blog.csdn.net/xeseo/article/details/17683049
  * @author Comsys-dengjun.zeng
  * @date 2016年6月30日 下午2:38:29
  *
 */
public class ExclaimBasicTopology {

	public static void main(String[] args) {
		TopologyBuilder builder = new TopologyBuilder();  
        
        builder.setSpout("spout", new RandomSpout(),5);  
        builder.setBolt("exclaim", new ExclaimBasicBolt()).shuffleGrouping("spout");  
        builder.setBolt("print", new PrintBolt(),3).shuffleGrouping("exclaim");  
  
        Config conf = new Config();  
        conf.setDebug(false);  
  
        if (args != null && args.length > 0) {  
            conf.setNumWorkers(3);  
  
            try {
				StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
			} catch (AlreadyAliveException e) {
				e.printStackTrace();
			} catch (InvalidTopologyException e) {
				e.printStackTrace();
			}  
        } else {  
  
            LocalCluster cluster = new LocalCluster();  
            cluster.submitTopology("test", conf, builder.createTopology());  
            Utils.sleep(100000);  
            cluster.killTopology("test");  
            cluster.shutdown();  
        }  

	}

}
